/**
 * 
 */
package com.eis.persist.dao.impl;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.eis.function.out.rule.StatusRuleResult;
import com.eis.persist.dao.FunctionDAO;
import com.eis.persist.dao.base.BaseDAO;
import com.eis.persist.dao.util.FunctionDataBatchPreparedStatementSetter;
import com.eis.persist.dao.util.FunctionSchemaBatchPreparedStatementSetter;
import com.eis.persist.dao.util.FunctionTempDataBatchPreparedStatementSetter;
import com.eis.persist.dao.util.FunctionTimeSetter;
import com.eis.persist.model.function.PersistDataFetchSync;
import com.eis.persist.model.function.PersistFunction;
import com.eis.persist.model.function.PersistFunctionDestInvokeInfo;
import com.eis.persist.model.function.PersistFunctionParameter;
import com.eis.persist.model.function.PersistFunctionSchema;
import com.eis.persist.model.function.PersistFunctionStatus;
import com.eis.persist.model.function.PersistFunctionStatusRule;
import com.eis.persist.model.function.PersistFunctionType;
import com.eis.thirdsys.connector.ThirdSysConnectorFactory;
import com.eis.thirdsys.connector.datatype.DataTypeList;
import com.eis.thirdsys.connector.datatype.bean.DataType;
import com.eis.thirdsys.connector.datatype.eissystype.MysqlDataType;

/**
 * @author panwei01
 *
 */
public class FunctionDAOImpl extends BaseDAO implements FunctionDAO {
	private static final Logger logger = LoggerFactory.getLogger(FunctionDAOImpl.class);
	private static final Map<String, Boolean> FUNCTION_UPDATE_LOCK = new HashMap<String, Boolean>(); 
	
	public FunctionDAOImpl(JdbcTemplate dataSource) {
		super(dataSource);
	}
	
	public FunctionDAOImpl(JdbcTemplate dataSource, String currentUserName) {
		super(dataSource, currentUserName);
	}

	public Map<Integer, PersistFunction> loadALlEnabledFunctions() {
		Map<Integer, PersistFunction> sapFunctions = new HashMap<Integer, PersistFunction>();
		// load all functions
		String sql = "select func.*,typee.typeName,typee.DisplayName,typee.description,bizsys.name as bizsysName," +
				"bizsys.connectorStaticsClassName," +
				"connconf.name as configName,connconf.configXml " +
				"from tb_function func " +
				"inner join tb_function_type typee on func.typeId=typee.id " +
				"inner join tb_biz_system bizsys on func.bizsysId=bizsys.id " +
				"left join tb_connector_config connconf on func.bizsysConnectorConfig=connconf.id " + 
				"where isActive=1";
		List<Map<String, Object>> allFunctions = dataSource.queryForList(sql);
		for (Map<String, Object> oneFunction : allFunctions) {
			sapFunctions.put((Integer)oneFunction.get("id"), new PersistFunction(oneFunction));
		}

		// load function schema
		String sql2 = "select fschema.* from tb_function func inner join tb_functionschema fschema on func.id=fschema.functionDBID" +
				" where func.isActive=1 order by fschema.id, fschema.functionDBID";
		List<Map<String, Object>> allFunctionSchemas = dataSource.queryForList(sql2);
		for (Map<String, Object> oneFunctionSchema : allFunctionSchemas) {
			PersistFunctionSchema schema = new PersistFunctionSchema(oneFunctionSchema);
			PersistFunction persistFunction = sapFunctions.get(schema.getFunctionDBID());
			persistFunction.getFunctionSchema().add(schema);
		}
		
		// load function parameters
		String sql3 = "select para.* from tb_function func inner join tb_functionparameter para on func.id=para.functionDBID" +
				" where func.isActive=1 order by para.id, para.functionDBID";
		List<Map<String, Object>> allFunctionParameters = dataSource.queryForList(sql3);
		for (Map<String, Object> oneFunctionPara : allFunctionParameters) {
			PersistFunctionParameter parameter = new PersistFunctionParameter(oneFunctionPara);
			PersistFunction persistFunction = sapFunctions.get(parameter.getFunctionDBID());
			persistFunction.getFunctionParameter().add(parameter);
		}
		
		return sapFunctions;
	}
	
	@Transactional(propagation=Propagation.REQUIRED)
	public void updateOrInsertFunctionSchema(PersistFunction persistFunction, ArrayList<HashMap<String, Object>> latestSchema, Date fetchTime) {
		if (latestSchema == null || latestSchema.size() == 0) return;
		
		DataTypeList functionDataTypeList;
		
		try {
			ThirdSysConnectorFactory thirdSysConnectorFactory = new ThirdSysConnectorFactory(persistFunction.getBizSystem().getConnectorStaticsClassName());
			functionDataTypeList = thirdSysConnectorFactory.newDataTypeList();
		} catch (Exception e) {
			logger.error("", e);
			return;
		}
		
		List<PersistFunctionSchema> originalSchema = persistFunction.getFunctionSchema();
		String sql = "insert into tb_functionschema(functionDBID,fieldName,fieldLength,fieldType,isExt,defaultDisplayName,createDate) " +
				"values(?,?,?,?,?,?,CURRENT_TIMESTAMP)";
		List<PersistFunctionSchema> newSchemas = new ArrayList<PersistFunctionSchema>();
		if (originalSchema == null || originalSchema.size() == 0) {
			String dataTableName = "tb_fdata_" + persistFunction.getFunctionID().toLowerCase();
			StringBuilder createDateTableSql = new StringBuilder("create table ");
				createDateTableSql.append(dataTableName);
				createDateTableSql.append("(eissysmainid INT AUTO_INCREMENT PRIMARY KEY,");
				createDateTableSql.append("functionDBID INT NOT NULL,");
				createDateTableSql.append("eissystimstp TIMESTAMP,");
				createDateTableSql.append("EISITEMSTATUS VARCHAR(32),");
			
			// sql for create status table
//			StringBuilder createStatusTableSql = new StringBuilder("create table tb_fdata_status_");
//			createStatusTableSql.append(persistFunction.getFunctionID().toLowerCase());
//			createStatusTableSql.append("(id INT AUTO_INCREMENT PRIMARY KEY,");
//			createStatusTableSql.append("eissysmainid INT NOT NULL,");
//			createStatusTableSql.append("eissystimstp TIMESTAMP,");
			
			for(HashMap<String, Object> oneSchema : latestSchema) {
				PersistFunctionSchema newSchema = new PersistFunctionSchema(persistFunction.getId(), oneSchema);
				newSchemas.add(newSchema);
				persistFunction.getFunctionSchema().add(newSchema); // update to function (without id)
				// append create table sql
				createDateTableSql.append(newSchema.getFieldName());
				
//				createStatusTableSql.append(newSchema.getFieldName());
				// data type
				DataType dataType = functionDataTypeList.valueOf(newSchema.getFieldType());
				createDateTableSql.append(" ").append(dataType.getMysqlDataType().name());
				
//				createStatusTableSql.append(" TINYINT,");
				if (MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
					createDateTableSql.append("(").append(
							newSchema.getFieldLength() == 0 ? 32 : newSchema.getFieldLength()).append(")");
				}
				createDateTableSql.append(",");
			}
			if (persistFunction.getDataUniqueKey() != null && persistFunction.getDataUniqueKey().length > 0) {
				createDateTableSql.append("UNIQUE KEY idx_tb_function_functionID (");
				for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
					String keyField = persistFunction.getDataUniqueKey()[i];
					createDateTableSql.append(keyField);
					if (i < persistFunction.getDataUniqueKey().length - 1) {
						createDateTableSql.append(",");
					}
				}
				createDateTableSql.append("),");
			}
			createDateTableSql.append("INDEX idx_tb_function_eissystimstp (eissystimstp),");
			createDateTableSql.append("FOREIGN KEY (functionDBID) REFERENCES tb_function(id))DEFAULT CHARSET=UTF8");
			
//			createStatusTableSql.append("INDEX idx_tb_function_status_eissystimstp (eissystimstp),");
//			createStatusTableSql.append("UNIQUE KEY (eissysmainid),");
//			createStatusTableSql.append("FOREIGN KEY (eissysmainid) REFERENCES " + dataTableName + "(eissysmainid))DEFAULT CHARSET=UTF8");
			// create PersistFunction Data table
			dataSource.execute(createDateTableSql.toString());
			// create PersistFunction Status table
//			dataSource.execute(createStatusTableSql.toString());
			
			StringBuilder createDataLogSql = new StringBuilder("create table tb_fdata_" + persistFunction.getFunctionID().toLowerCase() + "_log(");
			createDataLogSql.append("id INT AUTO_INCREMENT PRIMARY KEY,");
			createDataLogSql.append("dataId INT NOT NULL,");
			createDataLogSql.append("requestTimeMills BIGINT NOT NULL,");
			createDataLogSql.append("logDate TIMESTAMP DEFAULT CURRENT_TIMESTAMP,");
			createDataLogSql.append("FOREIGN KEY (dataId) REFERENCES tb_fdata_" + persistFunction.getFunctionID().toLowerCase() + "(eissysmainid),");
			createDataLogSql.append("index idx_datalog_requestTimeMills(requestTimeMills)");
			createDataLogSql.append(")DEFAULT CHARSET=UTF8");
			// create PersistFunction Data log table
			dataSource.execute(createDataLogSql.toString());
			
			// insert all
			FunctionSchemaBatchPreparedStatementSetter pss = new FunctionSchemaBatchPreparedStatementSetter(newSchemas);
			dataSource.batchUpdate(sql, pss);
		} else {
			// check difference and update
			Map<String, PersistFunctionSchema> originalFieldNameMap = new HashMap<String, PersistFunctionSchema>();
			for (PersistFunctionSchema oneField : persistFunction.getFunctionSchema()) {
				originalFieldNameMap.put(oneField.getFieldName(), oneField);
			}
			
			List<PersistFunctionSchema> newFields = new ArrayList<PersistFunctionSchema>();
			List<PersistFunctionSchema> updateTypeFields = new ArrayList<PersistFunctionSchema>();
			List<PersistFunctionSchema> updateLengthFields = new ArrayList<PersistFunctionSchema>();
			List<PersistFunctionSchema> removeFields = new ArrayList<PersistFunctionSchema>();
			
			Map<String, PersistFunctionSchema> newFieldNameMap = new HashMap<String, PersistFunctionSchema>();
			for (HashMap<String, Object> latest : latestSchema) {
				PersistFunctionSchema newSchema = new PersistFunctionSchema(persistFunction.getId(), latest);
				if (originalFieldNameMap.get(newSchema.getFieldName()) == null) {
					// add a new row
					newFields.add(newSchema);
				} else {
					PersistFunctionSchema os = originalFieldNameMap.get(newSchema.getFieldName());
					DataType dataType = functionDataTypeList.valueOf(newSchema.getFieldType());
					if (!os.getFieldType().equals(newSchema.getFieldType())) {
						updateTypeFields.add(newSchema);
					} else if (os.getFieldLength() != newSchema.getFieldLength() && MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
						updateLengthFields.add(newSchema);
					}
				}
				
				newSchemas.add(newSchema);
				newFieldNameMap.put(newSchema.getFieldName(), newSchema);
			}
			
			for (PersistFunctionSchema os : persistFunction.getFunctionSchema()) {
				if(newFieldNameMap.get(os.getFieldName()) == null) {
					removeFields.add(os);
				}
			}

			// have new fields, update the table
			if (newFields.size() > 0) {
				StringBuilder updateSchemaSql = new StringBuilder("ALTER TABLE tb_fdata_");
				updateSchemaSql.append(persistFunction.getFunctionID().toLowerCase());
				updateSchemaSql.append(" ");
				for (int i = 0; i < newFields.size(); i++) {
					PersistFunctionSchema oneNewField = newFields.get(i);
					// append create table sql
					updateSchemaSql.append(" ADD ");
					updateSchemaSql.append(oneNewField.getFieldName());
					// data type
					DataType dataType = functionDataTypeList.valueOf(oneNewField.getFieldType());
					updateSchemaSql.append(" ").append(dataType.getMysqlDataType());
					
					if (MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
						updateSchemaSql.append("(").append(oneNewField.getFieldLength()).append(")");
					}
					
					if (i < newFields.size() - 1) updateSchemaSql.append(",");
				}
				
				// add new fields
				dataSource.execute(updateSchemaSql.toString());
			}
			
			if (updateTypeFields.size() > 0) {
				StringBuilder dropSchemaSql = new StringBuilder("ALTER TABLE tb_fdata_");
				StringBuilder updateSchemaSql = new StringBuilder("ALTER TABLE tb_fdata_");
				
				dropSchemaSql.append(persistFunction.getFunctionID().toLowerCase());
				updateSchemaSql.append(persistFunction.getFunctionID().toLowerCase());
				
				dropSchemaSql.append(" ");
				updateSchemaSql.append(" ");
				
				for (int i = 0; i < updateTypeFields.size(); i++) {
					PersistFunctionSchema oneNewField = updateTypeFields.get(i);
					dropSchemaSql.append("DROP COLUMN ");
					dropSchemaSql.append(oneNewField.getFieldName());
					
					// append create table sql
					updateSchemaSql.append(" ADD ");
					updateSchemaSql.append(oneNewField.getFieldName());
					// data type
					DataType dataType = functionDataTypeList.valueOf(oneNewField.getFieldType());
					updateSchemaSql.append(" ").append(dataType.getMysqlDataType());
					
					if (MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
						updateSchemaSql.append("(").append(oneNewField.getFieldLength()).append(")");
					}
					
					if (i < newFields.size() - 1) {
						dropSchemaSql.append(",");
						updateSchemaSql.append(",");
						
					}
				}
				
				// drop original
				dataSource.execute(dropSchemaSql.toString());
				// add new fields
				dataSource.execute(updateSchemaSql.toString());
			}
			
			if (updateLengthFields.size() > 0) {
				StringBuilder updateSchemaSql = new StringBuilder("ALTER TABLE tb_fdata_");
				updateSchemaSql.append(persistFunction.getFunctionID().toLowerCase());
				updateSchemaSql.append(" ");
				for (int i = 0; i < updateLengthFields.size(); i++) {
					PersistFunctionSchema oneNewField = updateLengthFields.get(i);
					// append create table sql
					updateSchemaSql.append(" CHANGE COLUMN '");
					updateSchemaSql.append(oneNewField.getFieldName());
					updateSchemaSql.append("' '");
					updateSchemaSql.append(oneNewField.getFieldName());
					updateSchemaSql.append("'");
					// data type
					DataType dataType = functionDataTypeList.valueOf(oneNewField.getFieldType());
					updateSchemaSql.append(" ").append(dataType.getMysqlDataType());
					
					if (MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
						updateSchemaSql.append("(").append(oneNewField.getFieldLength()).append(")");
					}
					
					if (i < newFields.size() - 1) updateSchemaSql.append(",");
				}
				
				// add new fields
				dataSource.execute(updateSchemaSql.toString());
			}
			
			if (removeFields.size() > 0) {
				StringBuilder dropSchemaSql = new StringBuilder("ALTER TABLE tb_fdata_");
				
				dropSchemaSql.append(persistFunction.getFunctionID().toLowerCase());
				
				dropSchemaSql.append(" ");
				
				for (int i = 0; i < removeFields.size(); i++) {
					PersistFunctionSchema oneNewField = removeFields.get(i);
					dropSchemaSql.append("DROP COLUMN ");
					dropSchemaSql.append(oneNewField.getFieldName());

					if (i < newFields.size() - 1) {
						dropSchemaSql.append(",");
					}
				}
				
				// drop original
				dataSource.execute(dropSchemaSql.toString());
			}
			
		}
		
		String cleanSql = "delete from tb_functionschema where functionDBID=?";
		dataSource.update(cleanSql, persistFunction.getId());
		// insert to function schema
		FunctionSchemaBatchPreparedStatementSetter pss = new FunctionSchemaBatchPreparedStatementSetter(newSchemas);
		dataSource.batchUpdate(sql, pss);
	}
	
	public int batchSaveFunctionData(PersistFunction persistFunction, ArrayList<HashMap<String, Object>> functionData, Date date) throws Throwable {
		// add lock
		FUNCTION_UPDATE_LOCK.put(persistFunction.getFunctionID(), true);
		
		try {
			return batchSaveFunctionDataInternal(persistFunction, functionData, date);
		} catch (Throwable t) {
			throw t;
		} finally {
			// release lock
			FUNCTION_UPDATE_LOCK.put(persistFunction.getFunctionID(), false);
		}
	}
	
	public int batchSaveFunctionDataInternal(PersistFunction persistFunction, ArrayList<HashMap<String, Object>> functionData, Date date) throws Throwable {
		Date starttime = new Date(); // for log
		int recordCount = 0;
		// if the function is report function clean it up
		ArrayList<HashMap<String, Object>> functionsNeedTOBeupdated = new ArrayList<HashMap<String, Object>>();
		ArrayList<HashMap<String, Object>> functionsNeedTOBeInsert = new ArrayList<HashMap<String, Object>>();
		DataTypeList functionDataTypeList;
		try {
			ThirdSysConnectorFactory thirdSysConnectorFactory = new ThirdSysConnectorFactory(persistFunction.getBizSystem().getConnectorStaticsClassName());
			functionDataTypeList = thirdSysConnectorFactory.newDataTypeList();
		} catch (ClassNotFoundException e) {
			logger.error("", e);
			return 0;
		} catch (InstantiationException e) {
			logger.error("", e);
			return 0;
		} catch (IllegalAccessException e) {
			logger.error("", e);
			return 0;
		}
		
		logger.debug("loop in function: " + persistFunction.getFunctionID());
		if (PersistFunctionType.TYPE_REPORT.equalsIgnoreCase(persistFunction.getFunctionType().getTypeName())) {
			
			String deleteSql = "delete from tb_fdata_" + persistFunction.getFunctionID().toLowerCase();
			dataSource.execute(deleteSql);
			String resetIdSql = "ALTER TABLE tb_fdata_" + persistFunction.getFunctionID().toLowerCase() + " AUTO_INCREMENT=1";
			dataSource.execute(resetIdSql);
			functionsNeedTOBeInsert = functionData;
		} else {
			// check if function data duplicate
			Set<String> keySet = new HashSet<String>();
			logger.debug(persistFunction.getFunctionID() + " data rows: " + functionData.size());
			for(HashMap<String, Object> oneRow : functionData) {
				String key = "";
				for (String oneKey : persistFunction.getDataUniqueKey()) {
					key = key + "-" + String.valueOf(oneRow.get(oneKey));
				}
				
				if (keySet.contains(key)) {
					logger.warn(persistFunction.getFunctionID() + " duplicate key: " + key);
				} else {
					keySet.add(key);
				}
			}
			
			if (persistFunction.getDataUniqueKey() != null && persistFunction.getDataUniqueKey().length > 0) {
				// look for columns need to be updated
				// create a temp table to store all ids
				String tempTableName = "tmp_fdata_" + persistFunction.getFunctionID().toLowerCase();
				StringBuilder temTableSql = new StringBuilder("CREATE TEMPORARY TABLE " + tempTableName);
				temTableSql.append("(id INT AUTO_INCREMENT PRIMARY KEY,");
				
				// sql for insert ids into temp table 
				StringBuilder insertIdSql = new StringBuilder("insert into " + tempTableName + " (");
				StringBuilder insertIdValues = new StringBuilder("(");
				
				// sql for find the functions need to be updated
				StringBuilder seekUpdateSql = new StringBuilder("select data.eissysmainid,");
				
				List<PersistFunctionSchema> tempTableSchema = new ArrayList<PersistFunctionSchema>();
				
				for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
					String dataUniqueKey = persistFunction.getDataUniqueKey()[i];
					for (PersistFunctionSchema persistFunctionSchema : persistFunction.getFunctionSchema()) {
						if (persistFunctionSchema.getFieldName().equalsIgnoreCase(dataUniqueKey)) {
							temTableSql.append(persistFunctionSchema.getFieldName());
							insertIdSql.append(persistFunctionSchema.getFieldName());
							insertIdValues.append("?");
							seekUpdateSql.append("temp." + dataUniqueKey);
							// data type
							DataType dataType = functionDataTypeList.valueOf(persistFunctionSchema.getFieldType());
							temTableSql.append(" ").append(dataType.getMysqlDataType());
							if (MysqlDataType.VARCHAR == dataType.getMysqlDataType()) {
								temTableSql.append("(").append(persistFunctionSchema.getFieldLength()).append(")");
							}
							
							temTableSql.append(",");
							
							if (i < persistFunction.getDataUniqueKey().length - 1) {
								insertIdSql.append(",");
								insertIdValues.append(",");
								seekUpdateSql.append(",");
							}
							
							tempTableSchema.add(persistFunctionSchema);
							break;
						}
					}
				}
				temTableSql.append("UNIQUE KEY idx_tb_function_functionID (");
				for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
					String keyField = persistFunction.getDataUniqueKey()[i];
					temTableSql.append(keyField);
					if (i < persistFunction.getDataUniqueKey().length - 1) {
						temTableSql.append(",");
					}
				}
				temTableSql.append("))");
				insertIdValues.append(")");
				insertIdSql.append(") values ").append(insertIdValues);
				dataSource.execute(temTableSql.toString());
				
				// put all ids in
				FunctionTempDataBatchPreparedStatementSetter pss 
					= new FunctionTempDataBatchPreparedStatementSetter(functionData, tempTableSchema, functionDataTypeList);
				dataSource.batchUpdate(insertIdSql.toString(), pss);
				
				// look for data which need be updated
				seekUpdateSql.append(" from " + tempTableName + " temp inner join tb_fdata_" + persistFunction.getFunctionID().toLowerCase() + " data on ");
				for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
					String dataUniqueKey = persistFunction.getDataUniqueKey()[i];
					seekUpdateSql.append("temp." + dataUniqueKey + "=data." + dataUniqueKey);
					if (i < persistFunction.getDataUniqueKey().length - 1) {
						seekUpdateSql.append(" and ");
					}
				}
				List<Map<String, Object>> updateFunctionIds = dataSource.queryForList(seekUpdateSql.toString());
				logger.debug(persistFunction.getFunctionID() + " update function ids: " + updateFunctionIds.size());
				Map<String, Integer> updateFunctionIdsMap = new HashMap<String, Integer>(); // key=dataUniqueKeyValue1 + dataUniqueKeyValue1 + dataUniqueKeyValue1… value=eissysmainid
				for (Map<String, Object> oneId : updateFunctionIds) {
					String key = "";
					for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
						String dataUniqueKey = persistFunction.getDataUniqueKey()[i];
						key = key + oneId.get(dataUniqueKey).toString();
						if (i < persistFunction.getDataUniqueKey().length - 1) {
							key = key + "-";
						}
					}
					Integer value = (Integer)oneId.get("eissysmainid");
					updateFunctionIdsMap.put(key, value);
				}
				for (HashMap<String, Object> oneData : functionData) {
					String key = "";
					for (int i = 0; i < persistFunction.getDataUniqueKey().length; i++) {
						String dataUniqueKey = persistFunction.getDataUniqueKey()[i];
						key = key + oneData.get(dataUniqueKey).toString();
						if (i < persistFunction.getDataUniqueKey().length - 1) {
							key = key + "-";
						}
					}
					if (updateFunctionIdsMap.get(key) != null) {
						oneData.put("eissysmainid", updateFunctionIdsMap.get(key));
						functionsNeedTOBeupdated.add(oneData);
					} else {
						functionsNeedTOBeInsert.add(oneData);
					}
				}
				
				//drop temp table
				String dropTempSql = "drop TEMPORARY TABLE " + tempTableName;
				dataSource.execute(dropTempSql);
			}
		}

		StringBuilder insertSql = new StringBuilder("insert into tb_fdata_" + persistFunction.getFunctionID().toLowerCase());
		StringBuilder updateSql = new StringBuilder("update tb_fdata_" + persistFunction.getFunctionID().toLowerCase() + " set functionDBID=?,eissystimstp=?,");
		StringBuilder values = new StringBuilder(" values(?,?,");
		StringBuilder updateChanged = new StringBuilder(" and (");
		insertSql.append("(functionDBID,eissystimstp,");
		for (int i = 0; i < persistFunction.getFunctionSchema().size(); i++) {
			PersistFunctionSchema schema = persistFunction.getFunctionSchema().get(i);
			insertSql.append(schema.getFieldName());
			updateSql.append(schema.getFieldName() + "=?");
			values.append("?");
			updateChanged.append("(");
			updateChanged.append("(")
				.append("? is not null and (")
				.append(schema.getFieldName())
				.append(" is null or ")
				.append(schema.getFieldName())
				.append("<>?")
				.append(")) or (? is null and ")
				.append(schema.getFieldName())
				.append(" is not null))");
			if (i < persistFunction.getFunctionSchema().size() - 1) {
				insertSql.append(",");
				updateSql.append(",");
				values.append(",");
				updateChanged.append(" or ");
			}
		}
		insertSql.append(")");
		values.append(")");
		updateChanged.append(")");
		updateSql.append(" where eissysmainid=?");
		updateSql.append(updateChanged);
		
		insertSql.append(values);
		logger.debug("insert function ids: " + functionsNeedTOBeInsert.size());
		FunctionDataBatchPreparedStatementSetter insetPss 
			= new FunctionDataBatchPreparedStatementSetter(functionsNeedTOBeInsert, persistFunction.getFunctionSchema(), persistFunction.getId(), date, false, functionDataTypeList);
		FunctionDataBatchPreparedStatementSetter updatePss 
			= new FunctionDataBatchPreparedStatementSetter(functionsNeedTOBeupdated, persistFunction.getFunctionSchema(), persistFunction.getId(), date, true, functionDataTypeList);
		try {
			if (functionsNeedTOBeInsert.size() > 0)  {
				int[] inserted = dataSource.batchUpdate(insertSql.toString(), insetPss);
				logger.debug(persistFunction.getFunctionID() + " insert: " + inserted.length);
				recordCount += inserted.length;
			}
			if (functionsNeedTOBeupdated.size() > 0)  {
				int[] updated = dataSource.batchUpdate(updateSql.toString(), updatePss);
				logger.debug(persistFunction.getFunctionID() + " update: " + updated.length);
				recordCount += updated.length;
			}
			
		} catch (Throwable t) {
			logger.error("error for saving/updating function data, functionId: " 
					+ persistFunction.getFunctionID(), t);
			throw t;
		}
		
		logger.debug("finished save funtiondata start time: " + starttime.toString() + " end time: " + new Date().toString() + " functionid=" + persistFunction.getFunctionID());
		return recordCount;
	}
	
	public List<PersistFunctionSchema> getFunctionSchemaByFunctionId(String functionID) {
		String sql = "select * from tb_function func inner join tb_functionschema fschema on func.id=fschema.functionDBID" +
				" where func.functionID like '" + functionID + "'";
		List<Map<String, Object>> functionSchemas = dataSource.queryForList(sql);
		List<PersistFunctionSchema> results = new ArrayList<PersistFunctionSchema>();
		for (Map<String, Object> oneFunctionSchema : functionSchemas) {
			PersistFunctionSchema schema = new PersistFunctionSchema(oneFunctionSchema);
			results.add(schema);
		}
		return results;
	}
	
	public List<Map<String, Object>> getFunctionData(String functionID, Date date) {
		Date starttime = new Date(); // for log
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String sql = "select * from tb_fdata_" + functionID.toLowerCase() + " where eissystimstp>'" + format.format(date) + "'";
		if (FUNCTION_UPDATE_LOCK.get(functionID) != null && FUNCTION_UPDATE_LOCK.get(functionID)) {
			// data locked
			sql = sql + " and 1=2";
			logger.debug("finished load funtiondata start time: " + starttime.toString() + " end time: " + new Date().toString() + " functionid=" + functionID
					+ " function locked.");
		}
		List<Map<String, Object>> result = dataSource.queryForList(sql);
		logger.debug("finished load funtiondata start time: " + starttime.toString() + " end time: " + new Date().toString() + " functionid=" + functionID
				+ " total records: " + result.size());
		return result; 
	}

	public void updateFunctionDateFetchTime(Date date, String functionId) {
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String sql = "update tb_function set lastDataFetchTime='" + format.format(date) + "' where functionID=?";
		dataSource.update(sql, functionId);
	}

	@Override
	public PersistFunctionDestInvokeInfo getFunctionDestInvokeInfo(String functionId) {
		String sql = "select * from tb_functionDestInvokeInfo as fdinfo " +
				"inner join tb_function as func on func.functionId=fdinfo.functionId " +
				"where func.functionId=? and func.isActive>0";
		List<Map<String, Object>> result = dataSource.queryForList(sql, functionId);
		if (result != null && result.size() > 0) {
			PersistFunctionDestInvokeInfo po = new PersistFunctionDestInvokeInfo(result.get(0));
			return po;
		} else {
			// TODO: log an warning
			return null;
		}
	}

	@Override
	public int batchUpdateFunctionStatus(List<StatusRuleResult> ruleResult, String functionID) {
		// set all to null
		String sql = "update tb_fdata_" + functionID 
				+ " set eissystimstp=?,EISITEMSTATUS=?" 
				+ " where eissysmainid=?";
		
		FunctionTimeSetter setter = new FunctionTimeSetter(ruleResult);
		dataSource.batchUpdate(sql, setter);
		
		return dataSource.batchUpdate(sql, setter).length;
	}
	
	@Override
	public PersistFunctionStatusRule loadFunctionStatusRule(String functionId) {
		String sql = "select * from tb_statusrule where functionID=?";
		
		List<Map<String, Object>> result = dataSource.queryForList(sql, functionId);

		if (result != null && result.size() > 0) {
			PersistFunctionStatusRule persistFunctionStatusRule = new PersistFunctionStatusRule();
			persistFunctionStatusRule.setRulexml((String) result.get(0).get("rulexml"));
			persistFunctionStatusRule.setId((Integer) result.get(0).get("id"));
			persistFunctionStatusRule.setFunctionID((String) result.get(0).get("functionID"));
			persistFunctionStatusRule.setMaxRuleId((Integer) result.get(0).get("maxRuleId"));
			
			return persistFunctionStatusRule;
		} else {
			return null;
		}
	}

	@Override
	public boolean updateStatusRule(String functionId,
			String ruleXml, int maxRuleId) {
		int result;
		if (loadFunctionStatusRule(functionId) != null) {
			String sql = "update tb_statusrule set ruleXml=?,maxRuleId=?,updatedBy=? where functionID=?";
			result = dataSource.update(sql, ruleXml, maxRuleId, this.currentUserName, functionId);
		} else {
			String sql = "insert into tb_statusrule(rulexml,functionID,maxRuleId,createBy,createDate) values(?,?,?,?,CURRENT_TIMESTAMP)";
			result =  dataSource.update(sql, ruleXml, functionId, maxRuleId, this.currentUserName);
		}
		return result > 0;
	}

	@Override
	public List<PersistFunctionStatusRule> loadAllFunctionStatusRules() {
		String sql = "select * from tb_statusrule";
		
		List<Map<String, Object>> result = dataSource.queryForList(sql);

		if (result != null && result.size() > 0) {
			List<PersistFunctionStatusRule> returnList = new ArrayList<PersistFunctionStatusRule>(); 
			for (Map<String, Object> oneRow : result) {
				PersistFunctionStatusRule persistFunctionStatusRule = new PersistFunctionStatusRule();
				persistFunctionStatusRule.setRulexml((String) oneRow.get("rulexml"));
				persistFunctionStatusRule.setId((Integer) oneRow.get("id"));
				persistFunctionStatusRule.setFunctionID((String) oneRow.get("functionID"));
				persistFunctionStatusRule.setMaxRuleId((Integer) oneRow.get("maxRuleId"));
				
				returnList.add(persistFunctionStatusRule);
			}
			
			return returnList;
		} else {
			return null;
		}
	}

	@Override
	public List<PersistFunctionStatus> loadFunctionStatuses() {
		String sql = "select * from tb_functionStatus order by statusPriority desc";
		List<Map<String, Object>> result = dataSource.queryForList(sql);

		if (result != null && result.size() > 0) {
			List<PersistFunctionStatus> returnList = new ArrayList<PersistFunctionStatus>(); 
			for (Map<String, Object> oneRow : result) {
				PersistFunctionStatus persistFunctionStatus = new PersistFunctionStatus();
				persistFunctionStatus.setId((Integer) oneRow.get("id"));
				persistFunctionStatus.setName((String) oneRow.get("name"));
				persistFunctionStatus.setCustomizedDisplay((String) oneRow.get("customizedDisplay"));
				persistFunctionStatus.setDescription((String) oneRow.get("description"));
				persistFunctionStatus.setStatusPriority((Integer) oneRow.get("statusPriority"));
				
				returnList.add(persistFunctionStatus);
			}
			
			return returnList;
		} else {
			return null;
		}
	}

	@Override
	public PersistFunction getFunctionById(String functionId) {
		// load all functions
		String sql = "select func.*,typee.typeName,typee.DisplayName,typee.description,bizsys.name as bizsysName," +
				"bizsys.connectorStaticsClassName," +
				"connconf.name as configName,connconf.configXml " +
				"from tb_function func " +
				"inner join tb_function_type typee on func.typeId=typee.id " +
				"inner join tb_biz_system bizsys on func.bizsysId=bizsys.id " +
				"left join tb_connector_config connconf on func.bizsysConnectorConfig=connconf.id " + 
				"where isActive=1 and func.functionId=?";
		List<Map<String, Object>> function = dataSource.queryForList(sql, functionId);
		if (function != null && function.size() > 0) {
			PersistFunction pf = new PersistFunction(function.get(0));
			return pf;
		} else {
			return null;
		}
		
	}

	@Override
	public PersistDataFetchSync getPersistDataFetchSync(String currentRunnerId, int runduration, int faultUpdateTimeout) {
		Date currentRun = new Date();
		String loadSql = "select * from tb_data_fetch_sync order by id asc";
		
		// try to insert
		String insertSql = "insert into tb_data_fetch_sync(lastRun, currentRunnerId, uniqueKey) values (?, ?, 666)";
		int insertcount = 0;
		try {
			insertcount = dataSource.update(insertSql, currentRun, currentRunnerId);
		} catch (Throwable t) {
		}
		
		if (insertcount == 0) {
			// insert fail, try update
			String updateSql = "update tb_data_fetch_sync set lastRun=? where currentRunnerId like ?";
			int updatecount = 0;
			try {
				updatecount = dataSource.update(updateSql, currentRun, currentRunnerId);
			} catch (Throwable t) {
			}
			
			if (updatecount == 0) {
				// update fail, check if last run is normal
				Map<String, Object> result = dataSource.queryForMap(loadSql);
				PersistDataFetchSync persistDataFetchSync = new PersistDataFetchSync(result);
				
				Calendar cal = Calendar.getInstance();
				cal.setTime(persistDataFetchSync.getLastRun());
				
				cal.add(Calendar.MILLISECOND, runduration + faultUpdateTimeout);
				
				Calendar currentCal = Calendar.getInstance();
				currentCal.setTime(currentRun);
				if (cal.before(currentCal)) {
					// insert fail, try update
					String updateSql2 = "update tb_data_fetch_sync set lastRun=?, currentRunnerId=?";
					dataSource.update(updateSql2, currentRun, currentRunnerId);
					
					String localIp;
					try {
						localIp = InetAddress.getLocalHost().getHostAddress();
						logger.info(this.getClass().getName() + "::task runner server changed to " + localIp);
					} catch (UnknownHostException e) {
						logger.info(this.getClass().getName() + "::task runner server changed to " + currentRunnerId);
					}
					
				}
			}
		}
		
		Map<String, Object> result = dataSource.queryForMap(loadSql);
		PersistDataFetchSync persistDataFetchSync = new PersistDataFetchSync(result);

		return persistDataFetchSync;
	}
}
